#include "thread/thread_pool.hpp"

ThreadPool::ThreadPool(size_t thread_num)
    : _thread_num{thread_num}
{
}

ThreadPool::~ThreadPool()
{
    if (!_is_stop)
    {
        force_stop_gracefully();
    }
}

void ThreadPool::start()
{
    if (_is_started)
    {
        throw std::runtime_error("the thread pool already started...");
    }

    _is_started = true;
    _add_thread_unsafe(_thread_num);
}

void ThreadPool::add_task(std::function<void()> &&task, int priority)
{
    std::unique_lock<std::mutex> lock(_thread_lock);
    _task_queue.push({{std::move(task), priority}});
}

void ThreadPool::add_task_unsafe(std::function<void()> &&task, int priority)
{
    _task_queue.push({{std::move(task), priority}});
}

void ThreadPool::sync()
{
    std::for_each(_thread_list.begin(), _thread_list.end(),
                  [](std::thread &t)
                  { t.join(); });
}

void ThreadPool::force_stop_gracefully()
{
    _is_started = false;
    _is_stop = true;
    this->sync();
}

size_t ThreadPool::get_task_queue_size()
{
    return this->_task_queue.size();
}

void ThreadPool::_add_thread_unsafe(size_t thread_num)
{
    for (int i = 0; i < thread_num; i++)
    {
        auto execute_func = _get_execute_func();
        _thread_list.emplace_back(execute_func);
    }
}

void ThreadPool::add_thread_safe(size_t thread_num)
{
    std::unique_lock<std::mutex> lock(_thread_lock);
    _thread_num += thread_num;

    for (int i = 0; i < thread_num; i++)
    {
        auto execute_func = _get_execute_func();
        _thread_list.emplace_back(std::move(execute_func));
    }
}

std::function<void()> ThreadPool::_get_execute_func()
{
    return [&]()
    {
        while (true)
        {
            if (_is_stop)
            {
                return;
            }

            std::unique_lock<std::mutex> lock(_thread_lock);
            while (_task_queue.empty())
            {
                _cv.wait(lock, [&]
                         { return !_task_queue.empty() || _is_stop; });

                if (_is_stop)
                {
                    return;
                }
            }

            auto job = _task_queue.top();
            _task_queue.pop();
            lock.unlock();
            
            try
            {
                job._task._execute_task();
            }
            catch (const std::exception &e)
            {
                std::cerr << "Caught an exception in ThreadPool: " << e.what() << std::endl;
            }
            catch (...)
            {
                std::cerr << "Caught an unknown exception in ThreadPool." << std::endl;
            }
        }
    };
}

size_t ThreadPool::get_cur_thread_num() const
{
    return _thread_num;
}

void ThreadPool::force_stop()
{
    _is_started = false;
    if (_is_stop)
    {
        throw std::runtime_error("thread pool already shutdown!");
    }

    for (auto &thread : _thread_list)
    {
        thread.detach();
    }
}

ThreadPool &get_thread_pool(size_t thread_num)
{
    static ThreadPool pool{thread_num};
    return pool;
}

//
void test()
{
    std::cout << "start" << std::endl;

    // 获取线程池
    auto &pool = get_thread_pool(8);

    // 向线程池中添加100个任务
    for (int i = 0; i < 100; i++)
    {
        pool.add_task_unsafe(
            [=]
            {
                { // 休眠一秒模拟任务耗时
                    std::this_thread::sleep_for(std::chrono::seconds(1));
                    std::cout << i << std::endl;
                }
            },
            i == 56 ? 3 : 0); // 第56个任务调高其优先级
    }

    // 启动线程池
    pool.start();

    // 阻塞8秒后优雅停机
    std::this_thread::sleep_for(std::chrono::seconds(8));
    pool.force_stop_gracefully();
}